package com.atguigu.pro2

import java.net.URL
import java.util

import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * @description: 优化后的登录失败检测,2s后再报警存在漏洞,cep:Complex Event Processing,一个或多个由简单事件构成的事件流通过一定的规则匹配,然后输出用户想得到的数据,满足规则的复杂事件
 * 个体模式,组合模式,模式组 start.times(3).where(_behavior.startWith("fav))
 * @time: 2021/4/6 10:29
 * @author: baojinlong
 * */


object LoginFailWithCep {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 不影响输出顺序
    environment.setParallelism(1)
    // 有事件时间的都设置这个时间
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val resource: URL = getClass.getResource("/LoginLog.csv")
    //val inputStream: DataStream[String] = environment.readTextFile(resource.getPath)
    val inputStream: DataStream[String] = environment.readTextFile("src/main/resources/LoginLog.csv")
    // 转换成样例类行,并提起时间戳和watermark

    val loginEventStream: DataStream[LoginEvent] = inputStream
      .map(data => {
        val arr: Array[String] = data.split(",")
        LoginEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      // 3秒延时
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
        override def extractTimestamp(t: LoginEvent): Long = t.timestamp * 1000
      })
    // 定义匹配的模式,要求是一个登录失败事件后紧跟另一个登录失败事件,2s内连续出现两次出现登录错误,几次失败就定义几个流
    val failStr: String = "fail"
    val loginFailPattern: Pattern[LoginEvent, LoginEvent] = Pattern
      .begin[LoginEvent]("firstFail").where(_.eventType.equals(failStr))
      .next("secondFail").where(_.eventType.equals(failStr))
      .within(Time.seconds(2))

    // 将模式应用到数据流上,得到一个PatternStream
    val patternStream: PatternStream[LoginEvent] = CEP.pattern(loginEventStream.keyBy(_.userId), loginFailPattern)
    // 检出符合模式的数据流,需要调用select
    val loginFailWarningStream: DataStream[LoginFailWarning] = patternStream.select(new LoginFailEventMatch)
    loginFailWarningStream.print("loginFailWarningStream")
    environment.execute("loginFailWarningStreamTest")
  }
}


class LoginFailEventMatch extends PatternSelectFunction[LoginEvent, LoginFailWarning] {
  override def select(map: util.Map[String, util.List[LoginEvent]]): LoginFailWarning = {
    // 当前匹配到的事件序列,就保存在Map里
    val firstFailEvent: LoginEvent = map.get("firstFail").get(0)
    val secondFailEvent: LoginEvent = map.get("secondFail").get(0)
    LoginFailWarning(firstFailEvent.userId, firstFailEvent.timestamp, secondFailEvent.timestamp, "login fail")
  }
}